package com.mccken.rapid.common.current.queue.flusher;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventTranslatorOneArg;
import com.lmax.disruptor.ExceptionHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.WorkHandler;
import com.lmax.disruptor.WorkerPool;
import com.lmax.disruptor.dsl.ProducerType;

/**
 * 并行的Flusher多生产者多消费者工具类，基于disruptor的
 * 
 * @program: rapid_gateway
 * @description:
 * @author: mccken
 * @create: 2022-01-16 17:11
 **/
public class ParallelFlusher<E> implements Flusher<E> {
	
	private RingBuffer<Holder> ringBuffer;
	
	private final EventListener<E> eventListener;

	private final WorkerPool<Holder> workerPool;
	
	private final ExecutorService executorService;
	
	private final EventTranslatorOneArg<Holder, E> eventTranslator;
	
	/**
	 * 创建RingBuffer
	 * 通过ringBuffer创建一个屏障
	 * 创建多个消费者数组
	 * 构建多个消费者工作池
	 * 设置多个消费者的 sequence 序号，用于单独统计消费进度，并设置到ringbuffer中
	 * 启动workerPool
	 * 
	 * @param builder
	 */
	private ParallelFlusher(Builder<E> builder) {

		this.eventTranslator = new HolderEventTranslator();
		this.eventListener = builder.listener;
		this.executorService = Executors.newFixedThreadPool(builder.threads, 
				new ThreadFactoryBuilder().setNameFormat("ParallelFlusher-" + builder.namePrefix + "-pool-%d").build());

		RingBuffer<Holder> ringBuffer = RingBuffer.create(builder.producerType, new HolderEventFactory(), builder.bufferSize, builder.waitStrategy);
		SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
		WorkHandler<Holder>[] workHandlers = new WorkHandler[builder.threads];
		for (int i = 0; i < workHandlers.length; i++) {
			workHandlers[i] = new HolderWorkHandler();
		}

		this.workerPool = new WorkerPool<>(ringBuffer,
				sequenceBarrier,
				new HolderExceptionHandler(),
				workHandlers);
		
		ringBuffer.addGatingSequences(workerPool.getWorkerSequences());
	}
	
	public static class Builder<E> {
		private ProducerType producerType = ProducerType.MULTI;
		
		private int bufferSize = 16 * 1024;
		
		private int threads = 1;
		
		private String namePrefix = "";
		
		private WaitStrategy waitStrategy = new BlockingWaitStrategy();
		
		// 消费者监听
		private EventListener<E> listener;
		
		public Builder<E> setProducerType(ProducerType producerType) {
			Preconditions.checkNotNull(producerType);
			this.producerType = producerType;
			return this;
		}
		
		public Builder<E> setThreads(int threads) {
			Preconditions.checkArgument(threads > 0);
			this.threads = threads;
			return this;
		}
		
		public Builder<E> setBufferSize(int bufferSize) {
			Preconditions.checkArgument(Integer.bitCount(bufferSize) == 1);
			this.bufferSize = bufferSize;
			return this;
		}
		
		public Builder<E> setNamePrefix(String namePrefix) {
			Preconditions.checkNotNull(namePrefix);
			this.namePrefix = namePrefix;
			return this;
		}
		
		public Builder<E> setWaitStrategy(WaitStrategy waitStrategy) {
			Preconditions.checkNotNull(waitStrategy);
			this.waitStrategy = waitStrategy;
			return this;
		}
		
		public Builder<E> setEventListener(EventListener<E> listener) {
			Preconditions.checkNotNull(listener);
			this.listener = listener;
			return this;
		}
		
		public ParallelFlusher build() {
			return new ParallelFlusher<>(this);
		}
		
	}
	
	public interface EventListener<E> {
		
		void onEvent(E event) throws Exception;
		
		void onException(Throwable ex, long sequence, E event);
		
	}
	
	private class Holder {
		private E event;
		
		public void setValue(E event) {
			this.event = event;
		}
	}
	
	private class HolderEventFactory implements EventFactory<Holder>{

		@Override
		public Holder newInstance() {
			return new Holder();
		}
	}
	
	private class HolderWorkHandler implements WorkHandler<Holder> {
		@Override
		public void onEvent(Holder event) throws Exception {
			eventListener.onEvent(event.event);
			event.setValue(null);
		}
	}
	
	private class HolderExceptionHandler implements ExceptionHandler<Holder> {

		@Override
		public void handleEventException(Throwable throwable, long l, Holder event) {
			Holder holder = event;
			
			try {
				eventListener.onException(throwable, l, holder.event);
			} catch (Exception e) {
				// ignore
			} finally {
				holder.setValue(null);
			}
		}

		@Override
		public void handleOnStartException(Throwable throwable) {
			throw new UnsupportedOperationException();
		}

		@Override
		public void handleOnShutdownException(Throwable throwable) {
			throw new UnsupportedOperationException();
		}
	}
	
	private class HolderEventTranslator implements EventTranslatorOneArg<Holder,E> {

		@Override
		public void translateTo(Holder holder, long l, E e) {
			holder.setValue(e);
		}
	}

	private static <E> void process(EventListener<E> listener, Throwable e, E event) {
		listener.onException(e,-1, event);
	}
	
	private static <E> void process(EventListener<E> listener, Throwable e, E... events) {
		for (E event : events) {
			process(listener, e, event);
		}
	}

	@Override
	public void add(E event) {
		final RingBuffer<Holder> temp = ringBuffer;
		if (temp == null) {
			process(this.eventListener, new IllegalStateException("ParallelFlusher is closed"), event);
			return;
		}
		
		try {
			ringBuffer.publishEvent(eventTranslator, event);
		} catch (NullPointerException e) {
			process(this.eventListener, new IllegalStateException("ParallelFlusher is closed"), event);
		}
	}

	@Override
	public void add(E... events) {
		final RingBuffer<Holder> temp = ringBuffer;
		if (temp == null) {
			process(this.eventListener, new IllegalStateException("ParallelFlusher is closed"), events);
			return;
		}
		
		try {
			ringBuffer.publishEvents(this.eventTranslator, events);
		} catch (NullPointerException e) {
			process(this.eventListener, new IllegalStateException("ParallelFlusher is closed"), events);
		}
	}

	@Override
	public boolean tryAdd(E event) {
		final RingBuffer<Holder> temp = ringBuffer;
		if (temp == null) {
			return false;
		}

		try {
			return ringBuffer.tryPublishEvent(eventTranslator, event);
		} catch (NullPointerException e) {
			return false;
		}
	}

	@Override
	public boolean tryAdd(E... events) {
		final RingBuffer<Holder> temp = ringBuffer;
		if (temp == null) {
			return false;
		}

		try {
			return ringBuffer.tryPublishEvents(eventTranslator, events);
		} catch (NullPointerException e) {
			return false;
		}
	}

	@Override
	public boolean isShutdown() {
		return ringBuffer == null;
	}

	@Override
	public void start() {
		this.ringBuffer = workerPool.start(executorService);
	}

	@Override
	public void shutdown() {
		RingBuffer<Holder> temp = ringBuffer;
		ringBuffer = null;
		if (temp == null) {
			return;
		}
		if (workerPool != null) {
			workerPool.drainAndHalt();
		}
		
		if (executorService != null) {
			executorService.shutdown();
		}
	}
}
